package com.shujia.sql

import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Demo2DFApi {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      .master("local")
      .appName("api")
      .config("spark.sql.shuffle.partitions", 2)
      .getOrCreate()

    //导入隐式转换
    import spark.implicits._

    val student: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",") //默认是逗号分隔
      //按照数据的顺序指定表结构
      .schema("id STRING , name STRING , age INT , gender STrING , clazz STRING")
      .load("spark/data/students.txt")


    /**
      * show :查看数据
      *
      */

    student.show()

    /**
      * where : 过滤数据
      *
      */
    //字符串表达式
    student.where("age > 22").show()

    //使用 列表达式的方式
    student.where($"age" >= 22).show()


    /**
      * Dataset[Row]  和 DataFrame同一个东西
      *
      */

    val filterDS: Dataset[Row] = student.filter((row: Row) => {
      val age: Int = row.getAs[Int]("age")
      age >= 22
    })

    filterDS.show()

    /**
      * select : 选择数据
      *
      */

    student.select("name", "age").show()

    //选择的时候也可以处理数据
    student.select($"name", $"age" + 1 as "age").show()


    /**
      * group By: 分组
      * 分组之后必须借一个聚合函数
      *
      */

    student.groupBy("clazz").max("age").show()


    //导入sql中素有的函数
    import org.apache.spark.sql.functions._

    //季孙每个班级最大的年龄
    student
      .groupBy("clazz")
      .agg(max($"age") as "maxAge")
      .show()

    //计算每个班级平均年龄
    student.groupBy("clazz")
      .agg(avg("age") as "avgAge")
      .show()


    //统计班级性别的人数
    student.groupBy($"clazz", $"gender")
      .agg(countDistinct($"id") as "c")
      .show()


    /**
      *
      * join : 表关联
      *
      * 如果关联的列名一样直接指定关联列名
      *
      */


    val score: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",") //默认是逗号分隔
      //按照数据的顺序指定表结构
      .schema("sid STRING , cId STRING , sco DOUBLE")
      .load("spark/data/score.txt")


    val joinDF: DataFrame = student.join(score, $"id" === $"sid", "inner")

    //如果关联的列名一样直接指定关联列名
    //    val joinDF: DataFrame = student.join(score, "id")

    joinDF
      .groupBy("name")
      .agg(sum($"sco") as "sumSco")
      .show()

  }
}
